package com.xnyzc.lhy.order.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.xnyzc.lhy.common.component.config.security.RocketConfig;
import com.xnyzc.lhy.common.component.redis.RedisCommon;
import com.xnyzc.lhy.common.constant.MqConstant;
import com.xnyzc.lhy.common.constant.MqConsumerGroupConstant;
import com.xnyzc.lhy.common.constant.RedisPrefixConstant;
import com.xnyzc.lhy.common.entity.DiDiRv;
import com.xnyzc.lhy.common.exception.DiDiException;
import com.xnyzc.lhy.common.penum.EDiDiErrorCode;
import com.xnyzc.lhy.common.penum.order.EDiDiOrderStatus;
import com.xnyzc.lhy.common.util.CheckUtil;
import com.xnyzc.lhy.order.entity.order.OaCOrderPersonal;
import com.xnyzc.lhy.order.entity.param.order.OrderDetailParam;
import com.xnyzc.lhy.order.entity.param.synchronization.DriverSingleSynchronousParam;
import com.xnyzc.lhy.order.entity.result.order.OrderDetailResults;
import com.xnyzc.lhy.order.feign.didi.IDidiService;
import com.xnyzc.lhy.order.mapper.order.OaCOrderPersonalMapper;
import com.xnyzc.lhy.order.service.order.IOaDOrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * 司机抢单结果回调保护机制
 *
 * @author zhengxu
 */
@Slf4j
@Component
public class MqDidiSheetSyncConsumer {
    private DefaultMQPushConsumer consumer;

    @Autowired
    RocketConfig config;

    @Autowired
    RedisCommon redisCommon;

    @Value("${rocketMq.isStart}")
    private boolean isStart;

    @Autowired
    private IDidiService iDidiService;

    @Autowired
    private IOaDOrderService iOaDOrderService;

    @Autowired
    private OaCOrderPersonalMapper oaCOrderPersonalMapper;

    /**
     * 开启监听
     */
    @PostConstruct
    public void init() {
        try {
            if (!isStart) {
                return;
            }
            consumer = new DefaultMQPushConsumer(MqConsumerGroupConstant.DIDI_SHEET);
            consumer.setNamesrvAddr(config.getNameServerAddr());
            consumer.subscribe(config.getTopic(), MqConstant.DIDI_GRAB_SHEET_TAG);
            consumer.setConsumeMessageBatchMaxSize(1);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    MessageExt messageExt = msgs.get(0);
                    return MqDidiSheetSyncConsumer.this.dealBody(messageExt);
                }
            });
            consumer.start();
            log.info("---------司机抢单结果回调保护机制消费者启动完成-----o(*^▽^*)┛-------");
        } catch (Exception e) {
            log.error("司机抢单结果回调保护机制消费者初始化失败: {}", e);
            log.info("---------司机抢单结果回调保护机制消费者初始化失败----(# ﾟДﾟ)--------");
        }
    }

    /**
     * 处理业务逻辑
     *
     * @param messageExt
     * @return
     */
    public ConsumeConcurrentlyStatus dealBody(MessageExt messageExt) {
        try {
            //发送的消息
            String data = new String(messageExt.getBody());
            if (CheckUtil.strIsEmpty(data)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            OrderDetailParam orderDetailParam = JSON.parseObject(data, OrderDetailParam.class);
            int cnt = 2;
            while (true) {
                //延迟5秒查询订单状态
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    log.error(e.getMessage());
                    throw e;
                }
                // 如果redis中有key则跳出
                String securityKey = RedisPrefixConstant.DIDI_GRAB_SHEET_RESULT + orderDetailParam.getOrder_id();
                if (redisCommon.hashKey(securityKey)) {
                    redisCommon.remove(securityKey);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                cnt--;
                try {
                    // 调用查询订单详情
//                    DiDiRv orderDetail = iDidiService.getOrderDetail(orderDetailParam);
//                    Object dataObject = orderDetail.getData();
//                    String dataStr = JSON.toJSONString(dataObject);
//                    OrderDetailResults detailResults = JSONObject.parseObject(dataStr, OrderDetailResults.class);
//                    if (CheckUtil.objIsEmpty(detailResults)) {
//                        log.error("查询订单详情失败，抢单保护机制重查失败！");
//                        continue;
//                    }
                    // 抢单成功
                    log.error("抢单成功11111111111111！");
//                    if (CheckUtil.strIsNotEmpty(detailResults.getPassenger_phone()) && CheckUtil.strIsNotEmpty(detailResults.getPassenger_phone_suffix())) {
                        log.error("抢单成功11111111111111！");
//                        if (StringUtils.equals("101", detailResults.getStatus())) {

                            // 调用司机抢单同步接口
                            log.info("查询订单详情成功，准备调用司机抢单同步接口，通知司机已接单");
                            DriverSingleSynchronousParam driverSingleSynchronousParam = new DriverSingleSynchronousParam();
                            driverSingleSynchronousParam.setOpen_oid(orderDetailParam.getOpen_oid());
                            driverSingleSynchronousParam.setOrder_id(orderDetailParam.getOrder_id());
//                            driverSingleSynchronousParam.setPassenger_phone(orderDetailParam.getPassenger_phone());
//                            driverSingleSynchronousParam.setPassenger_phone_suffix(orderDetailParam.getPassenger_phone_suffix());
                            driverSingleSynchronousParam.setStrive_status("1");
                            iDidiService.driverSingleSynchronousFeign(driverSingleSynchronousParam);
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                        }
//                    }
                } catch (Exception e) {
                    log.error("保护机制查询订单详情失败");
                }
                //跳出循环状态
                if (cnt < 1) {
                    break;
                }
            }
            // 10秒之后还是查不到抢单信息，调用取消订单
            orderDetailParam.setCancel_reason("调用滴滴订单详情失败");
            // 查询订单
            if (CheckUtil.strIsEmpty(orderDetailParam.getOrder_id())) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            QueryWrapper<OaCOrderPersonal> queryWrapper = new QueryWrapper<>();
            queryWrapper.select(OaCOrderPersonal.DIDI_ORDER_STATUS, OaCOrderPersonal.VERSION, OaCOrderPersonal.DRIVER_ID, OaCOrderPersonal.PASSENGER_PHONE_SUFFIX);
            queryWrapper.eq(OaCOrderPersonal.ORDER_NO, orderDetailParam.getOrder_id());
            List<OaCOrderPersonal> oaCOrderPersonalList = oaCOrderPersonalMapper.selectList(queryWrapper);
            if (CheckUtil.collectionIsEmpty(oaCOrderPersonalList)) {
                throw DiDiException.create(EDiDiErrorCode.orderPersonalSelectFail);
            }
            OaCOrderPersonal oaCOrderPersonal = oaCOrderPersonalList.get(0);
            // 判断一下状态是否可以取消如果是0则可以取消
            if (StringUtils.equals(oaCOrderPersonal.getDidiOrderStatus().toString(), EDiDiOrderStatus.waitingResponse.getDidiType().toString())) {
                iOaDOrderService.updateOrderPersonalStatus(orderDetailParam);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error("司机抢单结果回调保护机制失效: {}", e);
            throw DiDiException.create(EDiDiErrorCode.didiSheetMqFail);
        }
    }
}
